package cn.doitedu.api;

import beans.Order;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/21
 * @Desc: 学大数据，上多易教育
 *    请从socket服务端口上读数据，数据是字符串:
 *        order_id,product_id,brand_id,金额
 *        o1,p1,华为,4000
 *        o1,p2,小米,3800
 *        o1,p3,华为,6000
 *        o2,p4,苹果,4000
 *        o2,p5,苹果,3800
 *        o2,p2,小米,4200
 *    每个订单中，金额最大的那笔商品  env.socketTextStream , map, keyBy , maxBy/reduce
 *
 **/
public class _03_Exercise_02_Reduce {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        DataStreamSource<String> stream = env.socketTextStream("localhost", 9999);

        stream.map(_03_Exercise_02_Reduce::strToOrder)
                .keyBy(Order::getOrderId)
                //.maxBy("amount")
                .reduce(new MaxReducer())
                .print();

        env.execute();
    }

    public static Order strToOrder(String line){
        String[] split = line.split(",");
        return new Order(split[0],split[1],split[2],Double.parseDouble(split[3]) );
    }


    public static class MaxReducer implements ReduceFunction<Order>{
        @Override
        public Order reduce(Order o1, Order o2) throws Exception {
            return o2.getAmount()> o1.getAmount()? o2:o1;
        }
    }


}
